package com.zhaosc.spark.core
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import scala.collection.immutable.List
import util.control.Breaks._
import scala.collection.mutable.SortedSet

/**
 * 获取每个月的最高温度的两条数据
 */
object TopNTemperature {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local")
      .setAppName("SecondSort");

    val sc = new SparkContext(conf);
    val weatherRdd = sc.textFile("weather.txt").cache();

    val top2 = weatherRdd.map(log => {
      val params = log.split("\t");

      val month = params(0).substring(0, 7);
      val wendu = params(1).toDouble;
      Tuple2(month, wendu);
    }).groupByKey().map(v => {
      val key = v._1;
      val it = v._2.iterator;
      val temperatures = new Array[Double](2);
      while (it.hasNext) {
        val temperature = it.next();
        breakable(for (i <- 0 until temperatures.length) {
          if (temperatures(i) == null) {
            temperatures(i) = temperature;
             	break;
          } else {
            if (temperature > temperatures(i)) {
              for (j <- (0 until temperatures.length).reverse; if j > i) {
                temperatures(j) = temperatures(j - 1);
              }
              temperatures(i) = temperature;
               break;
            }
          }
        })
      }
      Tuple2(key, temperatures)
    }).collect();

    top2.foreach(x => println(x._1 + "****" + x._2(0) + "----" + x._2(1)))

    sc.stop();
  }

}